package com.leo.biz.service.msgcenter;

import com.leo.biz.utils.ConfigUtils;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.springframework.stereotype.Service;
import java.util.Date;
import java.util.Properties;
import java.util.Random;

/**
 * Created by leo on 2016/11/16.
 */
@Service
public class MsgSendService {

    private Producer<String, String> producer;

    public MsgSendService()
    {
        init();
    }

    public void init()
    {
        Properties props = new Properties();
        props.put("metadata.broker.list", ConfigUtils.getMetadataBrokerList());
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
        props.put("request.required.acks", "0");
        props.put("partitioner.class", "com.leo.biz.service.msgcenter.MsgPartitioner");
        ProducerConfig config = new ProducerConfig(props);
        producer=new Producer<String, String>(config);
    }

    public void send(String topic,String msg)
    {
        if(producer==null)
        {
            init();
        }
        KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, msg);
        producer.send(data);
    }

    public void send(String topic,String key,String msg)
    {
        if(producer==null)
        {
            init();
        }
        KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, key,msg);
        producer.send(data);
    }

    public  void test(){

        Random rnd = new Random();

//        Properties props = new Properties();
//        props.put("metadata.broker.list", "localhost:9092");
//        props.put("serializer.class", "kafka.serializer.StringEncoder");
//        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
//        props.put("request.required.acks", "0");
//
//
//        ProducerConfig config = new ProducerConfig(props);

       // producer = new Producer<String, String>(config);

        for (long nEvents = 0; nEvents < 3; nEvents++) {
            long runtime = new Date().getTime();
            String ip = "192.168.2." + rnd.nextInt(255);
            String msg = runtime + ",www.example.com," + ip;
            KeyedMessage<String, String> data = new KeyedMessage<String, String>("mytopic", msg);
            producer.send(data);
        }
        producer.close();
    }
}
